package com.spx.chapter08;

import com.spx.chapter05.pojo.Event;
import com.spx.util.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * create by undeRdoG on  2022-05-04  13:51
 * 凡心所向，素履以往，生如逆旅，一苇以航。
 */
public class SplitStreamTest {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        SingleOutputStreamOperator<Event> sourceStream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                );

        /*
        * 分流操作，使用侧输出流进行分流，更加灵活，可以完全覆盖split算子的应用场景，因此split已被弃用
        * 使用 ProcessFunction的context上下文拿到侧输出流
        * */

        // 定义输出标签，不同的流的泛型都可以不同
        OutputTag<Tuple3<String,String,Long>> maryTag = new OutputTag<Tuple3<String,String,Long>>("maryTag"){};

        OutputTag<String> bobTag = new OutputTag<String>("bobTag"){};


        SingleOutputStreamOperator<Event> resultStream = sourceStream.process(new splitStream(maryTag,bobTag));


        resultStream.getSideOutput(maryTag).print("Mary");
        resultStream.getSideOutput(bobTag).print("Bob");
        resultStream.print("Main");

        env.execute();

    }


    /**
    *  这里的泛型。Out 相当于表示主流的数据类型
    * */
    public static class splitStream extends ProcessFunction<Event,Event> {

        // 定义输出标签，不同的流的泛型都可以不同
        OutputTag<Tuple3<String,String,Long>> maryTag;

        OutputTag<String> bobTag;

        public splitStream(OutputTag<Tuple3<String, String, Long>> maryTag, OutputTag<String> bobTag) {
            this.maryTag = maryTag;
            this.bobTag = bobTag;
        }

        @Override
        public void processElement(Event value, Context ctx, Collector<Event> out) throws Exception {
            if ("Mary".equalsIgnoreCase(value.user)){
                ctx.output(maryTag, Tuple3.of(value.user, value.url, value.timestamp));
            }else if ("Bob".equalsIgnoreCase(value.user)){
                ctx.output(bobTag, value.user + " " + value.url + " " + value.getClass());
            }else {
                // 主流中的内容
                out.collect(value);
            }
        }
    }
}
